
from __future__ import absolute_import
from __future__ import with_statement

__metaclass__ = type
__test__ = False


import os
import sys
import time
import logging
import threading
import traceback

from unittest import _WritelnDecorator

from nose.config import Config
from nose.core import TestProgram
from nose.util import isclass, odict
from nose.plugins import DefaultPluginManager

from fenton import script
from fenton import reloader
from fenton.console import colours
from fenton.logging import patch_getlogger

from fenton.script import Monitor, run_with_reloader

VERBOSITY=3

def once(ign):
    setup_logging()
    runtests(sys.argv[1], False)

def testconfig(conffile):
    config = Config()
    config.stream = sys.stderr # StringIO()
    config.plugins = DefaultPluginManager()
    config.loggingConfig = conffile
    #config.stopOnError = True
    return config

def runtests(conffile, exit=True):
    argv = ['--with-doctest']
    config = testconfig(conffile)
    runner = ConsoleRunner(config, VERBOSITY)

    print >>sys.stderr, "" # blank line
    tp = TestProgram(argv=argv, config=config, exit=exit, testRunner=runner)
    success = tp.success
    del tp
    return success


def setup_logging():
    patch_getlogger()
    logging.basicConfig()

def install_crash_dummy(pdb=False):
    try:
        import IPython.ultraTB
        f = IPython.ultraTB.FormattedTB(mode='Verbose', color_scheme='Linux', call_pdb=pdb)
    except:
        if not pdb:
            return
        def f(*args, **kw):
            import pdb
            pdb.set_trace()
    sys.excepthook = f


C = colours(hasattr(sys.stderr, 'fileno') and os.isatty(sys.stderr.fileno()))

got_gtk = False
got_others = False

try:
    import pygtk
except ImportError:
    # other GUIs?
    pass
else:
    if os.environ.get('DISPLAY'):
        pygtk.require("2.0")
        import gtk
        got_gtk = True

class ConsoleRunner:
    separator1 = '=' * 70
    separator2 = '-' * 70

    def __init__(self, config, verbosity, errorClasses={}):        
        self.errorClasses = errorClasses
        self.config = config
        self.stream = _WritelnDecorator(config.stream)
        self.dots = verbosity == 1
        self.show_tests = verbosity > 1
        self.show_descriptions = verbosity > 2
        self.failures = []
        self.errors = []
        self.testsRun = 0
        self.shouldStop = 0

    def __repr__(self):
        return "<%s%s run=%i errors=%i failures=%i>" % \
               (self.__class__.__module__, self.__class__.__name__,
                self.testsRun, len(self.errors),
                len(self.failures))

    # unittest TestResult API
    def startTest(self, test):
        if self.started is None:
            self.started = time.time()
        self.testsRun += 1
        if self.show_tests:
            self.stream.write(self.getDescription(test))
            self.stream.write(" ... ")

    def stopTest(self, test):
        "Called when the given test has been run"
        pass

    def stop(self):
        "Indicates that the tests should be aborted"
        self.shouldStop = True

    def addSuccess(self, test):
        if self.show_tests:
            self.stream.writeln("%sOK%s" % (C.Green, C.Normal))
        elif self.dots:
            self.stream.write('%s.%s' % (C.Green, C.Normal))

    def addError(self, test, err):
        """Overrides normal addError to add support for
        errorClasses. If the exception is a registered class, the
        error will be added to the list for that class, not errors.
        """
        ec, ev, tb = err
        exc_info = self._exc_info_to_string(err, test)
        for cls, (storage, label, isfail) in self.errorClasses.items():
            if isclass(ec) and issubclass(ec, cls):
                storage.append((test, exc_info))
                if self.show_tests:
                    self.stream.writeln(label)
                elif self.dots:
                    self.stream.write(label[:1])
                return
        self.errors.append((test, exc_info))
        cols = (C.LightRed, C.Normal)
        if self.show_tests:
            self.stream.writeln('%sERROR%s' % cols)
        elif self.dots:
            self.stream.write('%sE%s' % cols)

    def addFailure(self, test, err):
        self.failures.append((test, self._exc_info_to_string(err, test)))
        cols = (C.Red, C.Normal)
        if self.show_tests:
            self.stream.writeln("%sFAIL%s" % cols)
        elif self.dots:
            self.stream.write('%sF%s' % cols)

    def wasSuccessful(self):
        """Overrides to check that there are no errors in errorClasses
        lists that are marked as errors that should cause a run to
        fail.
        """
        if self.errors or self.failures:
            return False
        for cls in self.errorClasses.keys():
            storage, label, isfail = self.errorClasses[cls]
            if not isfail:
                continue
            if storage:
                return False
        return True

    def _exc_info_to_string(self, err, test):
        """Converts a sys.exc_info()-style tuple of values into a string."""
        exctype, value, tb = err
        # Skip test runner traceback levels
        while tb and self._is_relevant_tb_level(tb):
            tb = tb.tb_next
        if exctype is test.failureException:
            # Skip assert*() traceback levels
            length = self._count_relevant_tb_levels(tb)
            return ''.join(traceback.format_exception(exctype, value, tb, length))
        return ''.join(traceback.format_exception(exctype, value, tb))

    def _is_relevant_tb_level(self, tb):
        return tb.tb_frame.f_globals.has_key('__unittest')

    def _count_relevant_tb_levels(self, tb):
        length = 0
        while tb and not self._is_relevant_tb_level(tb):
            length += 1
            tb = tb.tb_next
        return length

    def run(self, test):
        # nose business
        wrapper = self.config.plugins.prepareTest(test)
        if wrapper is not None:
            test = wrapper

        # plugins can decorate or capture the output stream
        wrapped = self.config.plugins.setOutputStream(self.config.stream)
        if wrapped is not None:
            self.config.stream = wrapped
        
        self.started = time.time()
        test(self)
        self.report()
        self.config.plugins.finalize(self)
        #import pdb; pdb.set_trace()
        return self

    def report(self):
        finished = time.time()
        self.printErrors()
        self.printSummary(self.started, finished)

    def getDescription(self, test):
        if self.show_descriptions:
            return test.shortDescription() or str(test)
        else:
            return str(test)

    def printErrors(self):
        """Overrides to print all errorClasses errors as well.
        """
        # unittest
        if self.dots or self.show_tests:
            self.stream.writeln()
        self.printErrorList('%sERROR%s' % (C.LightRed, C.Normal), self.errors)
        self.printErrorList('%sFAIL%s' % (C.Red, C.Normal), self.failures)
        # nose
        for cls in self.errorClasses.keys():
            storage, label, isfail = self.errorClasses[cls]
            self.printErrorList(label, storage)
        self.config.plugins.report(self.stream)

    def printErrorList(self, flavour, errors):
        for test, err in errors:
            self.stream.writeln("%s%s%s" % (C.Normal, self.separator1, C.Normal))
            self.stream.writeln("%s: %s" % (flavour, str(test)))
            self.stream.writeln("%s%s%s" % (C.Normal, self.separator2, C.Normal))
            self.stream.writeln("%s" % err)

    def printSummary(self, start, stop):
        """Called by the test runner to print the final summary of test
        run results.
        """
        write = self.stream.write
        writeln = self.stream.writeln
        taken = float(stop - start)
        run = self.testsRun
        plural = run != 1 and "s" or ""
        
        writeln("%s%s%s" % (C.Normal, self.separator2, C.Normal))
        writeln("Ran %s test%s in %.3fs" % (run, plural, taken))
        writeln()
        if not self.wasSuccessful():
            write("%sFAILED%s (" % (C.Yellow, C.Normal))
            summary = odict()
            summary['%sfailures%s' % (C.Red, C.Normal)] = len(self.failures)
            summary['%serrors%s' % (C.LightRed, C.Normal)] = len(self.errors)
            for cls in self.errorClasses.keys():
                storage, label, isfail = self.errorClasses[cls]
                if not isfail:
                    continue
                summary[label] = len(storage)
            any = False
            for label, count in summary.items():
                if not count:
                    continue
                if any:
                    write(", ")
                write("%s=%s" % (label, count))
                any = True
            writeln(")")
        else:
            writeln("%sOK%s" % (C.Green, C.Normal))
        writeln()

def icon(name):
    path = os.path.dirname(os.path.abspath(__file__))
    return os.path.join(path, 'media', 'img', name)

class Icon:
    icons = {
        None: icon('grey.png'),
        True: icon('green.png'),
        False:icon('red.png')
    }

    def __init__(self, handler):
        self.handler = handler
        init = getattr(self, 'init', None)
        if init is not None:
            init()

    def show(self, status):
        icon = self.icons[status]
        self.change_icon(icon)

    def onclick(self, icon):
        self.handler()

    def change_icon(self, icon):
        print "showing " + icon

class GtkIcon(Icon):
    def init(self):
        icon = self.icons[None]
        self.icon = gtk.status_icon_new_from_file(icon)
        self.icon.connect('activate', self.onclick)
        gtk.gdk.threads_init()
        threading.Thread(target=gtk.main).start()

    def change_icon(self, img):
        self.icon.set_from_file(img)

if got_gtk:
    icon_class = GtkIcon
else:
    icon_class = Icon

suite = script.Suite(parser=script.suite_parser('test/config.ini'))

RELOAD_KEY = 'fenton_TEST_RELOADER'

@suite.builtin
def testrunner(self):
    if os.environ.get(RELOAD_KEY):
        def run():
            run_tests(self)
        return run_with_reloader(run, RELOAD_KEY)
    else:
        return run_monitor(self)

def run_tests(self):
    # reloader watches for modifications to loaded modules
    reloader.watch_file(self.config_file)
    reloader.watch_file(__file__)
    
    # patch nose to collect names of files
    # that might have syntax errors, so we can watch them too
    from nose.selector import Selector
    orig_wantFile = Selector.wantFile
    def new_wantFile(self, file):
        want = orig_wantFile(self, file)
        if want:
            reloader.watch_file(file)
        return want
    Selector.wantFile = new_wantFile
  
    config = testconfig(self.config_file)
    result = runtests(self.config_file, False)

    # print any logging
    import logging
    logs = (getattr(h, 'baseFilename', None) for h in logging.root.handlers)
    logs = [f for f in logs if f and os.path.getsize(f)]
    if logs:
        sys.stderr.write("Log:\n\n")
        for f in logs:
            sys.stderr.write(open(f).read())
    return int(result)


def run_monitor(self):
    import os, signal

    def onclick():
        os.kill(m.proc.pid, signal.SIGHUP)
    icon = icon_class(onclick)

    def report(arg):
        icon.show(bool(int(arg)))
    def finish():
        icon.show(None)

    m = Monitor(RELOAD_KEY, report, finish)
    return m.run()
    

